package com.imooc.log

import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * 使用Spark完成我们的数据清洗操作
 */
object SparkStatCleanJob {
  def main(args: Array[String]): Unit = {

    val spark=SparkSession
      .builder()
      .appName("SparkStatFormatJob")
      .config("spark.sql.parquet.compression.codec","gzip")
      .master("local[2]")
      .getOrCreate()

    val accessRDD = spark.sparkContext.textFile("file:///D:\\CodingFiles\\IdeaProjects\\ImoocSparkSQLProject\\data\\access.log")

//    accessRDD.take(10).foreach(println)

    //RDD ==> DF
    val rowRDD = accessRDD.map(x=>AccessConvertUtils.parseLog(x)).filter(x=>x.length>1)
    val accessDF = spark.createDataFrame(rowRDD, AccessConvertUtils.struct)
//    accessDF.printSchema()
//    accessDF.show(10)

    accessDF
      .coalesce(1)
      .write
      .format("parquet")
      .mode(SaveMode.Overwrite)
      .partitionBy("day")
      .save("file:///D:\\CodingFiles\\IdeaProjects\\ImoocSparkSQLProject\\data\\access_clean.parquet")

    spark.stop()

  }





}
